package com.baselib;

/**
 * script Name: Tunnel.java
 * @author 	Zhikun Cai @ Thomson Beijing office
 * @version 	1.0
 * function:	Socket tool for message communication.
 * 		Get an message object from the buffer.
 * date:	2008-09-17
 * update date: 2008-10-31 
 * update item: version:1.0
 *		Remove socket list. Close wrong socket immediately
 * 		because of 24*7 running usage. 
 */

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;


public class Tunnel {
	
	protected Message msg;
	protected Socket socket;
	protected BufferedReader socketInputReader;
	protected BufferedWriter socketOutputWriter;
	protected int msgHeadLen;
	protected int msgBodyLen;
	protected int serialNumber;
	protected int msgBeginPos;
	protected int msgBufSize;
	protected int reConnTimes;
	protected int waitSeconds;
	// Remove socket list "socketList" since version 1.1
	protected ArrayList<Socket> socketList;
	protected String msgBuf = null;
	protected String errorInfo = null;
	protected final int MAX_BUF_SIZE = 2*1024*1024;
	protected final int PROTOCOL_SIZE = 8;
	protected final int READ_SIZE = 8;

	
	/**
	 * Create a tunnel with socket,serial number, reconnnetion times and interval.
	 * @param socket
	 * @param serialNumber
	 * @param reConnTimes
	 * @param waitSeconds
	 * @throws IOException
	 */
	Tunnel(Socket socket, int serialNumber, int reConnTimes, int waitSeconds) throws IOException {
		msg = new Message();
		this.socket = socket;
		//BufferedWriter wrapped for effiency.
		socketInputReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
		socketOutputWriter = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
		msgHeadLen = 0;
		msgBodyLen = 0;
		this.serialNumber = serialNumber;
		msgBeginPos = 0;
		msgBufSize = 0;
		this.reConnTimes = reConnTimes;
		this.waitSeconds = waitSeconds;
		socketList = new ArrayList<Socket>();
		msgBuf = "";
		errorInfo = "";
	}
	
	
	/**
	 * 
	 * @param socket
	 * @throws IOException
	 */
	public Tunnel(Socket socket) throws IOException {
		this(socket, 0, 0, 0);
	}
	
	
	/**
	 * 
	 * @param socket
	 * @param serialNumber
	 * @throws IOException
	 */
	Tunnel(Socket socket, int serialNumber) throws IOException {
		this(socket, serialNumber, 0, 0);
	}
	
	Tunnel(Socket socket, int reConnTimes, int waitSeconds) throws IOException {
		this(socket, 0, reConnTimes, waitSeconds);
	}
	
	/**
	 * Free reconnected socket.
	 * @see java.lang.Object#finalize()
	 */
	protected void finalize() {
		/*
		 * Remove socket list "socketList" since version 1.1
		for (int i=0; i<socketList.size(); i++) {
			try { socketList.get(i).close();}
			catch (IOException e) {}
		}
		 */

		try {
			socket.close();
			socketInputReader.close();
			socketOutputWriter.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
	}
	
	
	/**
	 * Receive a message package from socket stream.
	 * @return Message as a object model. Or null when encoutering error.
	 */
	public Message receive() {
		int recvBytes = 0;
		char[] headBuf = new char[PROTOCOL_SIZE];
		char[] bodyBuf = new char[READ_SIZE];
		String recvSerialNumber;
		String recvBody;
		
		if (msgBufSize >= MAX_BUF_SIZE) {
			msgBuf = msgBuf.substring(msgBeginPos);
			msgBufSize = msgBuf.length();
			msgBeginPos = 0;
		}
		
		while (msgBufSize - msgBeginPos < PROTOCOL_SIZE) {
			System.out.println("get:"+msgBufSize+"\tBeg:"+PROTOCOL_SIZE);
			try {
				recvBytes = socketInputReader.read(headBuf, 0, PROTOCOL_SIZE);
				//System.out.println("get:"+new String(headBuf)+"R:"+recvBytes);
			} catch (IOException e) {
				errorInfo += "Tunnel reader IOException!\n";
				e.printStackTrace();
				return null;
			}
			if (recvBytes == -1) return null; //end of stream
			msgBuf += new String(headBuf, 0, recvBytes);
			msgBufSize = msgBuf.length();
		}
		
		msgHeadLen = Integer.parseInt(msgBuf.substring(msgBeginPos, msgBeginPos+PROTOCOL_SIZE/2));
		msgBodyLen = Integer.parseInt(msgBuf.substring(msgBeginPos+PROTOCOL_SIZE/2, msgBeginPos+PROTOCOL_SIZE));
		System.out.println("he");
		while (msgBufSize - msgBeginPos < msgHeadLen + msgBodyLen + PROTOCOL_SIZE) {
			try {
				recvBytes = socketInputReader.read(bodyBuf, 0, READ_SIZE);
			} catch (IOException e) {
				errorInfo += "Tunnel reader I/O exception!\n";
				return null;
			}
			if (recvBytes == -1) return null; //end of stream.
			msgBuf += new String(bodyBuf, 0, recvBytes);
			msgBufSize = msgBuf.length();
		}
		System.out.println("Get:body:"+msgBuf);
		recvSerialNumber = msgBuf.substring(msgBeginPos + PROTOCOL_SIZE, 
				msgBeginPos + PROTOCOL_SIZE + msgHeadLen);
		recvBody = msgBuf.substring(msgBeginPos + PROTOCOL_SIZE + msgHeadLen,
				msgBeginPos + PROTOCOL_SIZE + msgHeadLen + msgBodyLen);

		if (recvSerialNumber.equals("")) {
			msg.setSerialNumber(this.serialNumber);
			this.serialNumber += 1;
		}
		else {
			msg.setSerialNumber(Integer.parseInt(recvSerialNumber));
		}
		msg.setBody(recvBody);
		msgBeginPos += PROTOCOL_SIZE + msgHeadLen + msgBodyLen;
		
		return msg;
	}
	
	
	/**
	 * Send message through socket and reconnect when it is necessary.
	 * This function is not thread safe because of using global veriable: __msg
	 * @param msgBody
	 * @param serialNumber
	 * @return serinal number of sent message.
	 * @throws SocketException when reconnection is failed. 
	 */
	public int send(String msgBody, int serialNumber) throws SocketException {
		boolean reconnect = true;
		String data = null;
		
		if (serialNumber < 0) {
			serialNumber = this.serialNumber;
			this.serialNumber += 1;
			if (this.serialNumber == Integer.MAX_VALUE)
				this.serialNumber = 1;
		}
		
		msg.setBody(msgBody);
		msg.setSerialNumber(serialNumber);
		data = msg.buildMsgBuf();
		
		while (reconnect) {
			try {
				
				socketOutputWriter.write(data);
				socketOutputWriter.flush();
				reconnect = false;
			} catch(IOException ioe) {
				reconnect();
			}
		}
		return serialNumber;
	}
	
	public int send(String msgBody) throws SocketException {
		return send(msgBody, -1);
	}
	
	
	/**
	 * Get error info when user wish to see what happened.
	 * @return Error info string.
	 */
	public String getErrorInfo() {
		return errorInfo;
	}
	
	
	/**
	 * reconnect by initializing a new socket.
	 * @throws SocketException when reconnect for "reConnTimes" times that setted by user.
	 */
	protected void reconnect() throws SocketException {
		InetAddress addr = socket.getInetAddress();
		int port = socket.getPort();
		
		if (reConnTimes == 0) {
			errorInfo += "Socket error when (re)connecting "+addr+':'+port+"!\n";
			System.err.println("Socket error when (re)connecting "+addr+':'+port+"!");
			throw new SocketException("Socket error when (re)connecting "+addr+':'+port);
		}
		errorInfo += "Reconnect to "+addr+':'+port+" after "+waitSeconds+" second(s)!\n";
		System.err.println("Reconnect to "+addr+':'+port+" after "+waitSeconds+" second(s)!");
		try {
			Thread.sleep(waitSeconds*1000);
		} catch (InterruptedException e) {}
		/*
		 * Remove socket list "socketList" since version 1.1
		 * Close the bad socket immediately
		socketList.add(socket);
		 */
		try {
			socket.close();
		} catch(IOException e) {
			System.err.println("Close old socket to "+addr+':'+port+" failed!");
			errorInfo += "Close old socket to "+addr+':'+port+" failed!";
		}
		try {
			socket = new Socket(addr, port);
			// need to set timeout to new init socket
			socketInputReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
			socketOutputWriter = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
		} catch(IOException ioe) {
			errorInfo += "Init socket error when reconnecting!\n";
		}
		reConnTimes -= 1;
	}
}
